package top.codecrab.prodConsumer;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 阻塞队列版 生产者-消费者
 *
 * @author codecrab
 * @since 2021年07月01日 18:53
 */
public class BlockingQueueDemo {
    public static void main(String[] args) {
        Resource resource = new Resource(new ArrayBlockingQueue<>(10));

        new Thread(() -> {
            try {
                resource.producer();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "producer").start();

        new Thread(() -> {
            try {
                resource.consumer();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "consumer").start();

        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        resource.close();
        System.out.println("5秒，停止生产");
    }
}

class Resource {
    private volatile boolean flag = true;
    private final AtomicInteger atomicInteger = new AtomicInteger();
    private final BlockingQueue<String> blockingQueue;

    public Resource(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
        System.out.println(blockingQueue.getClass().getName());
    }

    public void producer() throws InterruptedException {
        String data = null;
        boolean offer = false;

        while (flag) {
            data = atomicInteger.incrementAndGet() + "";
            offer = blockingQueue.offer(data, 2, TimeUnit.SECONDS);
            if (offer) {
                System.out.println(Thread.currentThread().getName() + "\t 生产者生产商品成功，data：" + data);
            } else {
                System.out.println(Thread.currentThread().getName() + "\t 生产者生产商品失败，data：" + data);
            }
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println(Thread.currentThread().getName() + "\t生产者停止生产");
    }

    public void consumer() throws InterruptedException {
        String poll = null;

        while (flag) {
            poll = blockingQueue.poll(2, TimeUnit.SECONDS);
            if (poll == null || poll.equals("")) {
                flag = false;
                System.out.println(Thread.currentThread().getName() + "\t 超过两秒，消费者停止消费");
                System.out.println();
                System.out.println();
                return;
            }
            System.out.println(Thread.currentThread().getName() + "\t 消费者消费：" + poll);
        }
    }

    public void close() {
        this.flag = false;
    }
}
